package com.rizzo.analytics.worker.redis.task;

import com.google.common.base.Splitter;
import com.google.common.base.Stopwatch;
import com.google.common.collect.ImmutableMap;
import com.rizzo.analytics.domain.account.Account;
import com.rizzo.analytics.domain.trigger.Trigger;
import com.rizzo.analytics.worker.repo.AccountRepository;
import com.rizzo.infrastructure.domain.KeyType;
import com.rizzo.infrastructure.domain.TrendType;
import com.rizzo.infrastructure.performance.Monitor;
import com.rizzo.infrastructure.redis.config.RedisConfigManager;
import com.rizzo.infrastructure.redis.key.RedisKey;
import com.rizzo.infrastructure.redis.key.RedisKeyBuilder;
import com.rizzo.infrastructure.redis.timeseries.TimeStep;
import org.apache.commons.math3.stat.descriptive.moment.Mean;
import org.apache.commons.math3.stat.descriptive.moment.StandardDeviation;
import org.apache.commons.math3.stat.descriptive.rank.Max;
import org.apache.commons.math3.stat.descriptive.rank.Min;
import org.joda.time.DateTime;
import org.joda.time.Period;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.BoundZSetOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;

import java.text.DecimalFormat;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;

/**
 * This task will first convert the real-time data into "rotating" trend-data.
 * Supplementary we check if the user of the data has notifications set for the specific service.
 */
public class RedisTrendCalculator extends RedisTask {

	private static final Logger LOGGER = LoggerFactory.getLogger(RedisTrendCalculator.class);
	public static final String STANDARD_DEVIATION_ROUND_PATTERN = "#.##";

	private RedisConfigManager redisConfigManager;
	private AccountRepository accountRepository;

	public RedisTrendCalculator(RedisTemplate<RedisKey, String> redisKeyTemplate, AccountRepository accountRepository,
			RedisConfigManager redisConfigManager) {
		super(redisKeyTemplate, TimeStep.parsePeriod(redisConfigManager.getSystemConfig("workerLockExpiration")));
		this.accountRepository = accountRepository;
		this.redisConfigManager = redisConfigManager;
	}

	/**
	 * This method cycles over all PERFORMANCE keys of one application and if the timestep has expired (i.e. it is not real-time any more)
	 * calculates all the trend-data of the timestep (avg, min, max, dev) and stores it into the trend ZSet.
	 */
	@Scheduled(fixedRate = 20000)
	@Monitor(userBound = false)
	public void perform() {
		Stopwatch stopwatch = new Stopwatch().start();
		LOGGER.info("Calculating the trends!");
		RedisKey redisKeyLike = new RedisKeyBuilder().key(RedisKey.REDIS_WILDCARD).type(KeyType.PERFORMANCE)
				.instanceIndex(RedisKey.REDIS_WILDCARD).instanceName(RedisKey.REDIS_WILDCARD)
				.category(RedisKey.REDIS_WILDCARD).data(RedisKey.REDIS_WILDCARD).data(RedisKey.REDIS_WILDCARD).build();
		// TODO FIXME this could potentially be a lenghty operation to cycle over all performance keys...
		Set<RedisKey> keys = redisKeyTemplate.keys(redisKeyLike);
		for (RedisKey key : keys) {
			String application = key.getApplication();
			final RedisKey redisKey = new RedisKeyBuilder().key(application).type(KeyType.LOCK).data(key.getCategory()).build();
			if (acquireLock(redisKey)) {
				try {
					String timeStep = key.getSecondData();
					if (!TrendType.isTrendType(timeStep)) {
						long timeStepTimeStamp = Long.parseLong(timeStep);
						DateTime timeStepTime = new DateTime(timeStepTimeStamp);
						DateTime calculationLimitTime = getCalculationTimeLimit(redisKey.getFirstData(), application);
						if (timeStepTime.isBefore(calculationLimitTime)) {
							LOGGER.debug("Performance key with timestep: " + timeStepTime.toString() + " has expired and needs to be calculated.");
							Map<TrendType, Double> trendData = calculateTrendData(key);
							// we check if we have to generate notifications...
							generateNotifications(key, trendData);
							persistTrendData(trendData, key, timeStep);
							// now delete the rt-key
							redisKeyTemplate.opsForValue().getOperations().delete(key);
						} else {
							LOGGER.debug("Performance key with timestep: " + timeStepTime.toString() + " has not expired.");
						}
					}
				} finally {
					unlock(redisKey);
				}
			}
		}
		stopwatch.stop();
		LOGGER.info("Done calculating the trends in: " + stopwatch.elapsedMillis() + " millis!");
	}

	private DateTime getCalculationTimeLimit(String user, String application) {
		DateTime now = new DateTime();
		Period performanceTimeStep = TimeStep.parsePeriod(redisConfigManager.getUserConfig(user, application, "performanceTimeStep"));
		Period twoTimeStepsBack = performanceTimeStep.plus(performanceTimeStep);
		return TimeStep.dateTimeFloor(now, twoTimeStepsBack);
	}

	/**
	 * Calculates the trend-data from a RedisKey. Since the realtime data is partitioned into an appending value
	 * we have to decode the dataPoints from the lengthy string-value.
	 *
	 * @param redisKey the redisKey to calculate the trend data for
	 * @return the compressed real-time data
	 */
	private Map<TrendType, Double> calculateTrendData(RedisKey redisKey) {
		String timeStepCallValues = redisKeyTemplate.opsForValue().get(redisKey);
		Max max = new Max();
		Min min = new Min();
		Mean mean = new Mean();
		StandardDeviation deviation = new StandardDeviation();
		for (String dataPoint : Splitter.fixedLength(TimeStep.DATA_POINT_LENGTH).split(timeStepCallValues)) {
			double duration = Double.parseDouble(
					dataPoint.substring(TimeStep.DATA_POINT_SPLIT_LENGTH, TimeStep.DATA_POINT_LENGTH));
			max.increment(duration);
			min.increment(duration);
			mean.increment(duration);
			deviation.increment(duration);
		}
		DecimalFormat newFormat = new DecimalFormat(STANDARD_DEVIATION_ROUND_PATTERN);
		double roundedDeviation = Double.valueOf(newFormat.format(deviation.getResult()));
		double roundedMean = Math.round(mean.getResult());
		double hits = mean.getN();
		return ImmutableMap.of(TrendType.MAX, max.getResult(), TrendType.MIN, min.getResult(),
				TrendType.MEAN, roundedMean, TrendType.DEVIATION, roundedDeviation, TrendType.HITS, hits);
	}

	private void generateNotifications(RedisKey redisKey, Map<TrendType, Double> trendData) {
		Account account = accountRepository.findByUsername(redisKey.getFirstData());
		if (account != null) {
			LOGGER.info("Found account for user: " + redisKey.getFirstData());
			for (Trigger trigger : account.getTriggers()) {
				if (trigger.getService().equals(redisKey.getCategory())) {
					RedisKey notificationRedisKey = new RedisKeyBuilder().key(redisKey.getApplication()).type(KeyType.NOTIFICATION)
							.category(redisKey.getCategory()).data(redisKey.getFirstData()).data(trigger.getId().toString()).build();
					BoundZSetOperations<RedisKey, String> notificationOperations = redisKeyTemplate.boundZSetOps(notificationRedisKey);
					LOGGER.info("Found matching trigger for category: " + redisKey.getCategory());
					for (Map.Entry<TrendType, Double> trendTypeDoubleEntry : trendData.entrySet()) {
						if (trigger.getTresholdType().getPrefix().equals(trendTypeDoubleEntry.getKey().getPrefix())) {
							LOGGER.info("Found matching tresholdType: " + trigger.getTresholdType().getPrefix());
							if (matchesInstance(trigger, redisKey.getInstanceIndex(), redisKey.getInstanceName())) {
								LOGGER.info("Found matching instance: " + redisKey.getInstanceName() + " - " + redisKey.getInstanceIndex());
								Double trendDataValue = trendTypeDoubleEntry.getValue();
								LOGGER.info("TrendData value: " + trendDataValue);
								Double notificationValue = trigger.getTreshold();
								LOGGER.info("Notification value: " + notificationValue);
								if (trendDataValue >= notificationValue) {
									LOGGER.info("Generating trigger for service: "
											+ trigger.getService() + " - Type: " + trigger.getTresholdType().getPrefix());
									notificationOperations.add(redisKey.getSecondData(), trendDataValue);
									redisKeyTemplate.expire(notificationRedisKey, 24, TimeUnit.HOURS);
									// TODO send GCM to user to tell him there are notifications for his service call
								}
							}
						}
					}
				}
			}
		} else {
			LOGGER.debug("Found RedisKey for user: " + redisKey.getFirstData() + " but account could not be found!");
		}
	}

	private boolean matchesInstance(Trigger trigger, String instanceIndex, String instanceName) {
		boolean instanceIndexMatches = false;
		boolean instanceNameMatches = false;
		if (trigger.getInstanceIndex() == null) {
			instanceIndexMatches = true;
		} else if (trigger.getInstanceIndex().equals(instanceIndex)) {
			instanceIndexMatches = true;
		}
		if (trigger.getInstanceName() == null) {
			instanceNameMatches = true;
		} else if (trigger.getInstanceName().equals(instanceName)) {
			instanceNameMatches = true;
		}
		return instanceNameMatches && instanceIndexMatches;
	}

	/**
	 * TODO FIXME adding performance-data from past timeSteps that were already trend-calculated will overwrite that previously
	 * trend-calculated data in the RedisTrendCalculator should recalculate the newly found performance with the already existing
	 * trend-data.
	 *
	 * @param trendData
	 * @param redisKey
	 * @param timeStep
	 */
	private void persistTrendData(Map<TrendType, Double> trendData, RedisKey redisKey, String timeStep) {
		for (Map.Entry<TrendType, Double> trendTypeDoubleEntry : trendData.entrySet()) {
			RedisKey trendKey = new RedisKeyBuilder().key(redisKey.getApplication()).instanceName(redisKey.getInstanceName())
					.instanceIndex(redisKey.getInstanceIndex()).type(KeyType.PERFORMANCE)
					.category(redisKey.getCategory()).data(redisKey.getFirstData()).data(trendTypeDoubleEntry.getKey().getPrefix()).build();
			redisKeyTemplate.boundZSetOps(trendKey).add(timeStep, trendTypeDoubleEntry.getValue());
		}
	}

}
